package com.dongrj.framework.starter.log.send;

import com.dongrj.framework.core.utils.JsonUtils;
import com.dongrj.framework.starter.log.model.ELKModel;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @author dongrongjun
 * @date 2020/06/18
 */
@Slf4j
public class KafkaElkSend implements ElkSend {

    private KafkaProducer<String, String> kafkaProducer;

    private static volatile KafkaElkSend instance;

    private KafkaElkSend() {
        log.info("KafkaElkSend正在初始化!");
        Properties properties = new Properties();
//        String kafkaAddress = "kafka-0.kafka-headless.logging.svc.cluster.local:9092,kafka-1.kafka-headless.logging.svc.cluster.local:9092,kafka-2.kafka-headless.logging.svc.cluster.local:9092";
        String kafkaAddress = "152.136.142.244:31001,152.136.142.244:31002,152.136.142.244:31003";
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaAddress);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.ACKS_CONFIG, "0");
        try {
            this.kafkaProducer = new KafkaProducer<>(properties);
        } catch (Exception e) {
            log.error("KafkaElkSend初始化失败,程序退出!", e);
            System.exit(-1);
        }
    }


    public static KafkaElkSend getInstance() {
        if (instance == null) {
            synchronized (KafkaElkSend.class) {
                if (instance == null) {
                    instance = new KafkaElkSend();
                }
            }
        }
        return instance;
    }

    @Override
    public void send(ELKModel elkModel) {
        String elk = JsonUtils.toJson(elkModel);
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>("elk-topic", elk);
        kafkaProducer.send(producerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if (e != null) {
                    System.out.println(e);
                }
            }
        });
    }
}
